package com.yeyouluo.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class CreateHBaseTable {
    public static final String TABLE_NAME = "friend";

    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;


    public static void main(String[] args) throws IOException {
        CreateHBaseTable createHBaseTable = new CreateHBaseTable();
        createHBaseTable.init();
        createHBaseTable.createTable();
        createHBaseTable.addFriendRelationShip("uid1", "uid2");
        createHBaseTable.addFriendRelationShip("uid1", "uid5");
        createHBaseTable.addFriendRelationShip("uid1", "uid7");
        createHBaseTable.addFriendRelationShip("uid2", "uid3");
        createHBaseTable.addFriendRelationShip("uid2", "uid6");
        createHBaseTable.addFriendRelationShip("uid2", "uid10");
        createHBaseTable.destroy();
    }

    public void createTable() throws IOException {
        admin = (HBaseAdmin) conn.getAdmin();
        // 创建表描述器
        HTableDescriptor teacher = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
        // 设置列族描述器
        teacher.addFamily(new HColumnDescriptor("friends"));
        // 执行创建操作
        admin.createTable(teacher);
        System.out.println("表[" + TABLE_NAME + "]创建成功");
    }

    public void addFriendRelationShip(String uidSrc, String uidTarget) throws IOException {
        // 获取table对象
        Table table = conn.getTable(TableName.valueOf(TABLE_NAME));

        // 准备put对象
        List<Put> putList = new ArrayList<>(2);
        Put putUid1 = new Put(Bytes.toBytes(uidSrc));
        putUid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(uidTarget), Bytes.toBytes(uidTarget));

        Put putUid2 = new Put(Bytes.toBytes(uidTarget));
        putUid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(uidSrc), Bytes.toBytes(uidSrc));

        putList.add(putUid1);
        putList.add(putUid2);

        // 插入数据
        table.put(putList);
        System.out.println("插入双向好友关系[" + uidSrc + "<-->" + uidTarget + "]数据成功");
    }

    @Deprecated
    public void addFriendRelationShip() throws IOException {
        // 获取table对象
        Table table = conn.getTable(TableName.valueOf(TABLE_NAME));

        // 准备put对象
        List<Put> putList = new ArrayList<>(16);
        Put putUid1 = new Put(Bytes.toBytes("uid1"));
        putUid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        putUid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid5"), Bytes.toBytes("uid5"));
        putUid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid7"), Bytes.toBytes("uid7"));

        Put putUid2 = new Put(Bytes.toBytes("uid2"));
        putUid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"), Bytes.toBytes("uid1"));
        putUid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"), Bytes.toBytes("uid3"));
        putUid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid6"), Bytes.toBytes("uid6"));
        putUid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid10"), Bytes.toBytes("uid10"));


        putList.add(putUid1);
        putList.add(putUid2);

        // 插入数据
        table.put(putList);
        System.out.println("插入好友关系数据成功");
    }


    public void init() throws IOException {
        // 获取配置
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        // 获取hbase连接
        conn = ConnectionFactory.createConnection(conf);
    }

    public void destroy() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
